package com.dc.rxjava;

import android.util.Log;

import java.util.concurrent.TimeUnit;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Predicate;

/**
 * 过滤操作符
 *
 * @author 冰柠檬dc
 */
public class FilterOperatorUtils {

    private static final String TAG = "RxJava";

    /**
     * filter操作符：通过一定逻辑来过滤被观察者发送的事件，如果返回 true 则会发送事件，否则不会发送。
     * TODO 注意和条件操作符takeWhile、takeUntil、skipWhile、skipUntil区别
     *
     * @return Disposable
     */
    public static Disposable filter() {
        return Observable.just(1, 4, 3, 5, 2, 6).filter(new Predicate<Integer>() {
            @Override
            public boolean test(Integer integer) throws Exception {
                return integer <= 4;
            }
        }).subscribe(integer -> {
            Log.e(TAG, "onNext:" + integer);
        }, throwable -> {
            Log.e(TAG, "onError");
        }, () -> {
            Log.e(TAG, "onComplete");
        }, onSubscribe -> {
            Log.e(TAG, "onSubscribe");
        });
        // 打印结果：
        // onSubscribe
        // onNext:1
        // onNext:4
        // onNext:3
        // onNext:2
        // onComplete
    }

    /**
     * ofType操作符：可以过滤不符合该类型事件
     *
     * @return Disposable
     */
    public static Disposable ofType() {
        return Observable.just(1, 2, 3, "dc").ofType(String.class).subscribe(s -> {
            Log.e(TAG, "onNext:" + s);
        }, throwable -> {
            Log.e(TAG, "onError");
        }, () -> {
            Log.e(TAG, "onComplete");
        }, onSubscribe -> {
            Log.e(TAG, "onSubscribe");
        });
        // 打印结果：
        // onSubscribe
        // onNext:dc
        // onComplete
    }

    /**
     * skip操作符：跳过正序某些事件，count 代表跳过事件的数量
     * skipLast操作符：用来跳过正序的最后面的事件
     *
     * @return Disposable
     */
    public static Disposable skip() {
        return Observable.just(1, 2, 3).skip(1).subscribe(s -> {
            Log.e(TAG, "onNext:" + s);
        }, throwable -> {
            Log.e(TAG, "onError");
        }, () -> {
            Log.e(TAG, "onComplete");
        }, onSubscribe -> {
            Log.e(TAG, "onSubscribe");
        });
        // 打印结果：skip
        // onSubscribe
        // onNext:2
        // onNext:3
        // onComplete

        // 打印结果：skipLast
        // onSubscribe
        // onNext:1
        // onNext:2
        // onComplete
    }

    /**
     * distinct操作符：过滤事件序列中的重复事件。
     *
     * @return Disposable
     */
    public static Disposable distinct() {
        return Observable.just(1, 2, 3, 3, 2, 1).distinct().subscribe(i -> {
            Log.e(TAG, "onNext:" + i);
        }, throwable -> {
            Log.e(TAG, "onError");
        }, () -> {
            Log.e(TAG, "onComplete");
        }, onSubscribe -> {
            Log.e(TAG, "onSubscribe");
        });
        // 打印结果：
        // onSubscribe
        // onNext:1
        // onNext:2
        // onNext:3
        // onComplete
    }

    /**
     * distinctUntilChanged操作符：过滤掉连续重复的事件
     *
     * @return Disposable
     */
    public static Disposable distinctUntilChanged() {
        return Observable.just(1, 2, 3, 3, 2, 1).distinctUntilChanged().subscribe(i -> {
            Log.e(TAG, "onNext:" + i);
        }, throwable -> {
            Log.e(TAG, "onError");
        }, () -> {
            Log.e(TAG, "onComplete");
        }, onSubscribe -> {
            Log.e(TAG, "onSubscribe");
        });
        // 打印结果：
        // onSubscribe
        // onNext:1
        // onNext:2
        // onNext:3
        // onNext:2
        // onNext:1
        // onComplete
    }

    /**
     * take操作符：控制观察者接收的事件的数量。
     * takeLast操作符：接受事件序列的后面几个事件
     *
     * @return Disposable
     */
    public static Disposable take() {
        return Observable.just(1, 2, 3, 4, 5).take(2).subscribe(i -> {
            Log.e(TAG, "onNext:" + i);
        }, throwable -> {
            Log.e(TAG, "onError");
        }, () -> {
            Log.e(TAG, "onComplete");
        }, onSubscribe -> {
            Log.e(TAG, "onSubscribe");
        });
        // 打印结果：take
        // onSubscribe
        // onNext:1
        // onNext:2
        // onComplete

        // 打印结果：takeLast
        // onSubscribe
        // onNext:4
        // onNext:5
        // onComplete
    }

    /**
     * debounce操作符：如果两件事件发送的时间间隔小于设定的时间间隔则前一件事件就不会发送给观察者。
     * throttleWithTimeout() 与此方法的作用一样，
     *
     * @return Disposable
     */
    public static Disposable debounce() {
        return Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                e.onNext(1);
                Thread.sleep(900);
                e.onNext(2);
            }
        }).debounce(1, TimeUnit.SECONDS).subscribe(i -> {
            Log.e(TAG, "onNext:" + i);
        }, throwable -> {
            Log.e(TAG, "onError");
        }, () -> {
            Log.e(TAG, "onComplete");
        }, onSubscribe -> {
            Log.e(TAG, "onSubscribe");
        });
        // 打印结果：
        // onSubscribe
        // onNext:2
        // onComplete
    }

    /**
     * firstElement操作符：取事件序列的第一个元素
     * lastElement操作符：取事件序列的最后一个元素。
     *
     * @return Disposable
     */
    public static Disposable firstElement() {
        return Observable.just(1, 2, 3).firstElement().subscribe(i -> {
            Log.e(TAG, "onNext:" + i);
        }, throwable -> {
            Log.e(TAG, "onError");
        }, () -> {
            Log.e(TAG, "onComplete");
        });
        // 打印结果：firstElement
        // onNext:1

        // 打印结果：lastElement
        // onNext:3
    }

    /**
     * elementAt操作符：elementAt() 可以指定取出事件序列中事件，但是输入的 index 超出事件序列的总数的话就不会出现任何结果。
     * 这种情况下，你想发出异常信息的话就用 elementAtOrError() 。
     *
     * @return Disposable
     */
    public static Disposable elementAt() {
        return Observable.just(1, 2, 3).elementAt(1).subscribe(i -> {
            Log.e(TAG, "onNext:" + i);
        }, throwable -> {
            Log.e(TAG, "onError");
        }, () -> {
            Log.e(TAG, "onComplete");
        });
        // 打印结果：
        // onNext:2
    }
}
